package com.we.mexico.flink.utils;

import org.apache.flink.api.java.utils.ParameterTool;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;

public class WeCdcPropertyReader {
    private static ParameterTool param;
    public static Logger LOG = LoggerFactory.getLogger(WeCdcPropertyReader.class);

    public static WeCdcPropertyReader init(String propFilePath) {
        try {
            param = getReader(propFilePath);
        } catch (Exception e) {
            LOG.error("Exception: " + e.getMessage());
            e.printStackTrace();
        }
        return new WeCdcPropertyReader();
    }

    public static ParameterTool getReader(String filePath) throws IOException {
        InputStream resourceAsStream =
                WeCdcPropertyReader.class.getClassLoader().getResourceAsStream(filePath);
        return ParameterTool.fromPropertiesFile(resourceAsStream);
    }

    public String getHost() {
        return param.get("host");
    }

    public int getPort() {
        return param.getInt("port");
    }

    public String getUserName() {
        return param.get("username");
    }

    public String getPassword() {
        return param.get("password");
    }

    public String getDatabase() {
        return param.get("database");
    }

    public String getTableName() {
        return param.get("tablename");
    }

    public String getServerId() {
        return param.get("serverid");
    }

    public int getSplitNum() {
        return param.getInt("splitnum");
    }

    public String getKfkSinkAdd() {
        return param.get("kfksinadd");
    }

    public String getKfkTopicPrefix() {
        return param.get("kfktopicprefix");
    }

    public String getKfkTopicSuffix() {
        return param.get("kfktopicsuffix");
    }

    public String getKeySeriaSchemaClm() {
        return param.get("keyseriaschemaclm");
    }

    public String getRocksDBBackendUrl() {
        return param.get("rocksDBBackendURL");
    }

    public static void main(String[] args) {
        WeCdcPropertyReader reader = WeCdcPropertyReader.init("cdcpipeline/Hhl01Source.properties");
        String tablelist = "";
        String database = reader.getDatabase();
        String tableName = reader.getTableName();
        StringBuilder stringBuilder = new StringBuilder(10);

        if (tableName.contains(",")) {
            String[] splits = tableName.split(",");
            for (int i = 0; i < splits.length; i++) {
                stringBuilder.append(database);
                stringBuilder.append(".");
                stringBuilder.append(splits[i]);
                if (i < splits.length - 1) {
                    stringBuilder.append(",");
                }
            }
            for (String split : splits) {
                tablelist += database + "." + split + ",";
            }
            tablelist = tablelist.substring(0, tablelist.length() - 1);
        } else {
            stringBuilder.append(database);
            stringBuilder.append(".");
            stringBuilder.append(tableName);
            tablelist += database + "." + tableName;
        }
        System.out.println(stringBuilder);
    }
}
